# Copyright (c) 2018 Presto Labs Pte. Ltd.
# Author: xguo

import enum
import functools
import hashlib
import json
import logging
import typing
import uuid

import requests
import tabulate
import tornado.gen
import tornado.httpclient
from recordclass import recordclass

import coin.flow.subscriber as flow_subscriber
import coin.flow.topic_map as topic_map
import coin.proto.coin_market_enums_pb2 as coin_enum
import coin.proto.coin_order_enums_pb2 as coin_order_pb2

from coin.base.config import Config
from coin.base.datetime_util import to_datetime
from coin.base.timestamp import get_timestamp
from coin.exchange.base.order_fill_util import FillManager
from coin.exchange.base.order_gateway import BaseOrderGatewayInfoMutator
from coin.exchange.base.order_gateway_ready import SpotOrderGatewayReady
from coin.exchange.base.rest.task_scheduler import Scheduler, wrap_async_task
from coin.exchange.base.rest_order_gateway_base import PrivateExchangeMessage
from coin.exchange.binance.enum_types import BinanceOrderSide, BinanceOrderType, BinanceExecutionType
from coin.exchange.binance.enum_types import BinanceOrderStatus
from coin.exchange.binance.kr_rest.currency import BinanceCurrency
from coin.exchange.binance.kr_rest.native_private_client import BinanceNativePrivateClient
from coin.exchange.binance.kr_rest.product import BinanceProduct
from coin.exchange.binance.ws.private_subscriber import BinancePrivateSubscriber
from coin.exchange.util.rate_limit import RateLimiter
from coin.exchange.util.rest_rate_limit import RestRateLimiter
from coin.proto.coin_order_gateway_pb2 import OrderGatewayConfig, OrderEvent
from coin.proto.coin_query_pb2 import CurrencyBalance, AccountBalance
from coin.strategy.hk_hard_arb.og_util import print_working_orders
from coin.exchange.base.rest_order_gateway_base import (
    RestOrderGateway,
    exchange_message_handler,
    create_order_internal,
)
from coin.exchange.base.order_gateway import (Order,
                                              OrderSide,
                                              OrderType,
                                              OrderGatewayStatus,
                                              assert_orders_equivalent)


def create_order_from_execution_report(message, order_id, timestamp):
  """
  {
  "e": "executionReport",        // Event type
  "E": 1499405658658,            // Event time
  "s": "ETHBTC",                 // Symbol
  "c": "CLIENT_ORDER_ID",        // Client order ID
  "S": "BUY",                    // Side
  "o": "LIMIT",                  // Order type
  "f": "GTC",                    // Time in force
  "q": "1.00000000",             // Order quantity
  "p": "0.10264410",             // Order price
  "P": "0.00000000",             // Stop price
  "F": "0.00000000",             // Iceberg quantity
  "g": -1,                       // Ignore
  "C": "null",                   // Original client order ID; This is the ID of the order being canceled
  "x": "NEW",                    // Current execution type
  "X": "NEW",                    // Current order status
  "r": "NONE",                   // Order reject reason; will be an error code.
  "i": 4293153,                  // Order ID
  "l": "0.00000000",             // Last executed quantity
  "z": "0.00000000",             // Cumulative filled quantity
  "L": "0.00000000",             // Last executed price
  "n": "0",                      // Commission amount
  "N": null,                     // Commission asset
  "T": 1499405658657,            // Transaction time
  "t": -1,                       // Trade ID
  "I": 8641984,                  // Ignore
  "w": true,                     // Is the order working? Stops will have
  "m": false,                    // Is this trade the maker side?
  "M": false,                    // Ignore
  "O": 1499405658657,            // Order creation time
  "Z": "0.00000000",             // Cumulative quote asset transacted quantity
  "Y": "0.00000000",             // Last quote asset transacted quantity (i.e. lastPrice * lastQty)
  "Q": "0.00000000",             // Quote Order Qty
  }
  """
  field_name_map = {
      "e": "event_type",
      "E": "event_time",
      "s": "symbol",
      "c": "client_order_id",
      "S": "order_side",
      "o": "order_type",
      "f": "time_in_force",
      "q": "quantity",
      "p": "price",
      "P": "stop_price",
      "F": "iceberg_quantity",
      "g": "ignore_0",
      "C": "original_client_order_id",
      "x": "execution_type",
      "X": "order_status",
      "r": "order_reject_reason",
      "i": "order_id",
      "l": "qty_last_fill",
      "z": "qty_accum_fill",
      "L": "price_last_fill",
      "n": "commission_amount",
      "N": "commission_asset",
      "T": "transaction_time",
      "t": "trade_id",
      "I": "ignore_1",
      "w": "is_working",
      "m": "is_maker",
      "M": "ignore_2",
      "O": "order_creation_time",
      "Z": "amount_accum_fill",
      "Y": "amount_last_fill",
      "Q": "quote_order_qty",
  }
  logger = logging.getLogger(__name__)

  if list(field_name_map.keys()) != list(message.keys()):
    logger.warning('Execution report message format changed!')

  message = {field_name_map[key]: message[key] for key in field_name_map}

  native_symbol = message['symbol']
  product = BinanceProduct.FromStr(native_symbol)

  order_side = {'BUY': OrderSide.BUY, 'SELL': OrderSide.SELL}[message['order_side']]
  # BinanceOrderType.LIMIT_MAKER = OrderType.LIMIT + post_only
  order_type = {'LIMIT': OrderType.LIMIT, 'LIMIT_MAKER': OrderType.LIMIT}[message['order_type']]

  price_orig = float(message['price'])
  qty_orig = float(message['quantity'])
  exchange_order_id = str(message['order_id'])
  qty_last_fill = float(message['qty_last_fill'])
  qty_accum_fill = float(message['qty_accum_fill'])
  price_last_fill = float(message['price_last_fill'])

  if qty_accum_fill > 1.e-10:
    price_avg_fill = float(message['amount_accum_fill']) / qty_accum_fill
  elif qty_accum_fill < 0:
    logger.error('Insane qty_accum_fill: %s', qty_accum_fill)
    qty_accum_fill = 0
    price_avg_fill = 0
  else:
    qty_accum_fill = 0
    price_avg_fill = 0

  fill = FillManager(
      price_orig=price_orig,
      price_last_fill=price_last_fill,
      price_avg_fill=price_avg_fill,
      qty_orig=qty_orig,
      qty_accum_fill=qty_accum_fill,
      qty_last_fill=qty_last_fill,
      timestamp_last_fill=timestamp,
  )
  order = Order(order_id=order_id,
                product=product,
                price=price_orig,
                qty=qty_orig,
                order_side=order_side,
                order_type=order_type,
                tag=None,
                accepted=True,
                cancel_sent=False,
                internal=create_order_internal(timestamp=timestamp,
                                               exchange_order_id=exchange_order_id,
                                               fill=fill))
  return order


def parse_fill_info_in_submit_resp(msg):
  assert 'fills' in msg
  native_symbol = msg['symbol']
  total_cost = 0
  qty_accum = 0
  trade_ids = []
  for fill_info in msg['fills']:
    qty = float(fill_info['qty'])
    cost = float(fill_info['price']) * qty
    qty_accum += qty
    total_cost += cost
    trade_ids.append('%s_%s' % (native_symbol, fill_info['tradeId']))
  return total_cost / qty_accum, qty_accum, trade_ids


def create_order_from_open_order_message(msg, timestamp, order_id):
  logger = logging.getLogger(__name__)
  native_symbol = msg['symbol']
  exchange_order_id = str(msg['orderId'])
  product = BinanceProduct.FromStrNativeProduct(native_symbol)
  price_orig = float(msg['price'])
  qty_orig = float(msg['origQty'])
  qty_accum_fill = float(msg['executedQty'])
  amount_accum_fill = float(msg['cummulativeQuoteQty'])

  if qty_accum_fill > 1.e-10:
    price_avg_fill = amount_accum_fill / qty_accum_fill
  elif qty_accum_fill < 0:
    logger.error('Insane qty_accum_fill: %s', qty_accum_fill)
    qty_accum_fill = 0
    price_avg_fill = 0
  else:
    qty_accum_fill = 0
    price_avg_fill = 0

  order_type = OrderType.LIMIT
  order_side = OrderSide.BUY if msg['side'] == 'BUY' else OrderSide.SELL

  fill = FillManager(price_orig=price_orig,
                     qty_orig=qty_orig,
                     price_last_fill=None,
                     qty_last_fill=None,
                     price_avg_fill=price_avg_fill,
                     qty_accum_fill=qty_accum_fill,
                     timestamp_last_fill=timestamp)

  order = Order(order_id=order_id,
                product=product,
                price=price_orig,
                qty=qty_orig,
                order_side=order_side,
                order_type=order_type,
                tag=None,
                accepted=True,
                cancel_sent=False,
                internal=create_order_internal(timestamp=timestamp,
                                               exchange_order_id=exchange_order_id,
                                               fill=fill))
  return order


class BinanceOrderGatewayInfo(BaseOrderGatewayInfoMutator):
  CurrencyType = BinanceCurrency
  ProductType = BinanceProduct
  BalanceType = float
  PositionType = None

  def __init__(self,
               currencies: typing.List[BinanceCurrency],
               products: typing.List[BinanceProduct]):
    super().__init__(currencies, products)
    self._ws_update_time = None

  def update_balance_rest(self, message):
    if (self._ws_update_time is not None and get_timestamp() - self._ws_update_time < 2e9):
      return

    for balance in message['balances']:
      currency = BinanceCurrency.FromStrNativeCurrencyNothrow(balance['asset'])
      if currency is None:
        continue
      available = float(balance['free'])
      hold = float(balance['locked'])
      total = available + hold
      assert available >= 0.0
      assert hold >= 0.0
      assert total >= 0.0
      currency_balance = CurrencyBalance(
          currency=currency.currency,
          currency_native=currency.native_currency,
          total=total,
          available=available,
          hold=hold,
      )
      self.set_balance(currency, currency_balance.total, currency_balance)

  def update_balance_ws(self, message):
    self._ws_update_time = get_timestamp()
    for balance in message['B']:
      currency = BinanceCurrency.FromStrNativeCurrencyNothrow(balance['a'])  # asset
      if currency is None:
        continue
      available = float(balance['f'])  # free
      hold = float(balance['l'])  # locked
      total = available + hold
      assert available >= 0.0
      assert hold >= 0.0
      assert total >= 0.0
      currency_balance = CurrencyBalance(
          currency=currency.currency,
          currency_native=currency.native_currency,
          total=total,
          available=available,
          hold=hold,
      )
      self.set_balance(currency, currency_balance.total, currency_balance)

  def balance_as_proto(self):
    return AccountBalance(exchange='Binance',
                          market_type='Spot',
                          each_balance=self.get_notNone_balances_internal())


OrderStats = recordclass('OrderStats',
                         [
                             'num_submitted',
                             'num_cancelled',
                             'num_submitted_success',
                             'num_cancelled_success',
                             'num_fills',
                             'disabled',
                             'disabled_timestamp',
                             'conversion_ratio',
                         ])


# For Binance's rate limiting rule, please refer to
# https://support.binance.com/hc/en-us/articles/360004492232-API-Frequently-Asked-Questions-FAQ
class BinanceOrderGateway(RestOrderGateway):
  _exchange = 'Binance'
  _market_type = 'Spot'
  _api_version = None

  OGInfoType = BinanceOrderGatewayInfo

  class _MsgType(enum.IntEnum):
    REST_ACCOUNT_BALANCE = 0
    WS_ACCOUNT_BALANCE = 1
    OPEN_ORDERS = 2
    ORDER_UPDATE = 3
    ACCOUNT_STATUS = 4
    SYSTEM_STATUS = 5
    WS_ACCOUNT_POSITION = 6

  def __init__(self,
               currencies: typing.List[BinanceCurrency],
               products: typing.List[BinanceProduct],
               config: Config,
               og_config=None,
               logger=None):

    super().__init__(currencies, products, config, og_config, logger)
    self._og_ready = SpotOrderGatewayReady()
    self._og_ready.set_status(OrderGatewayStatus.INITIALIZING)
    self._exchange_info = None

    self._private_client = None
    self._private_subscriber = None
    self._scheduler = None

    self._ioloop = None
    self._flow_sub = None
    self._publisher = None
    self._trade_ids = set()

    default_og_config = OrderGatewayConfig(account_query_period=2,
                                           orders_snapshot_query_period=4,
                                           order_rate_limit=10,
                                           order_rate_limit_per=1,
                                           hit_rate_limit_cooldown_time=300)
    default_og_config.MergeFrom(self._og_config)
    self._og_config = default_og_config
    self._og_logger.exchange_type = coin_enum.Binance

    self._rest_rate_limiter = RestRateLimiter(delay=self._og_config.hit_rate_limit_cooldown_time)
    self._order_rate_limiter = RateLimiter(self._og_config.order_rate_limit,
                                           self._og_config.order_rate_limit_per)
    self._og_logger.exchange_type = coin_enum.Binance

    self._uuid = 'presto' + hashlib.md5(str(
        uuid.uuid1()).encode()).hexdigest()[:16]  # for client order id

    self._order_stats = {}
    for product in self._og_info.products:
      self._order_stats[product] = OrderStats(
          num_submitted=0,
          num_cancelled=0,
          num_submitted_success=0,
          num_cancelled_success=0,
          num_fills=0,
          disabled=False,
          disabled_timestamp=None,
          conversion_ratio=0,
      )

  def _to_client_order_id(self, order):
    client_order_id = '%s_%s' % (self._uuid, order.order_id)
    return client_order_id

  def _parse_client_order_id(self, client_order_id):
    idx = client_order_id.find('_')
    if idx == -1:
      return None, None

    og_uuid = client_order_id[:idx]
    if og_uuid != self._uuid:
      self._logger.info('External order %s, self_og_id=%s', client_order_id, self._uuid)
      return None, None

    order_id = client_order_id[idx + 1:]
    return uuid, order_id

  def _disable_og(self):
    self._scheduler.stop()
    self._private_subscriber.stop()
    self._og_ready.reset()

  def _handle_error_response(self, response, extra_params=None):
    if response.status_code == requests.codes.unauthorized:  # 401
      # {"code": -2015, "msg": "Invalid API-key, IP, or permissions for action."}
      self._logger.error('Conversion ratio is too low!')
      extra_params = extra_params or {}
      order = extra_params.get('order')
      if order is not None:
        self._order_stats[order.product].disabled = True
        self._order_stats[order.product].disabled_timestamp = get_timestamp()
      self._rest_rate_limiter.set_error(status_code=response.status_code)
    elif response.status_code in (
        requests.codes.teapot, requests.codes.too_many_requests):  # (418, 420) rate limit exceeded
      self._logger.error('(%s) Hit rate limit, slow down!', response.status_code)
      self._on_hit_order_rate_limit()
    elif response.status_code == requests.codes.forbidden:  # 403:
      self._logger.error('Violate WAF rule!')
      self._on_hit_order_rate_limit()
    else:
      self._logger.error('Submit response error: %s, %s', response.status_code, response.content)

    task = wrap_async_task(
        self._private_client.query_account_status,
        msg_type=self._MsgType.ACCOUNT_STATUS,
        publisher=self._publisher,
    )
    self._scheduler.run_once(task)

  def start(self, queue, ioloop, flow_sub=None):
    api_host = self._og_config.rest_root_url \
      if self._og_config.HasField('rest_root_url') else None
    self._private_client = BinanceNativePrivateClient(key_file=self._config.key_file,
                                                      api_host=api_host,
                                                      use_async=True)

    topic_id = topic_map.search_topic_id('binance_rest_private_1')[0][1]
    self._ioloop = ioloop
    self._publisher = queue.get_writer(topic_id=topic_id)
    self._scheduler = Scheduler(ioloop=ioloop, logger=self._logger)
    self._prepare_private_subscriber()
    self._install_periodic_query()

    # Subscribe to flow for feed.
    self._flow_sub = flow_sub or flow_subscriber.from_queue(queue)
    self._flow_sub.subscribe('binance_rest_private_1', self.on_private_exchange_msg)
    self._get_exchange_info()

  def _prepare_private_subscriber(self, start=False):
    if self._private_subscriber is not None:
      self._private_subscriber.stop()

    self._private_subscriber = BinancePrivateSubscriber(
        self._private_client,
        self._ioloop,
        self._notification_callback,
        functools.partial(self._prepare_private_subscriber, start=True),
    )

    if start:
      self._private_subscriber.start()

  def submittable(self):
    return self.is_ready() and self._order_rate_limiter.check(
    ) and self._rest_rate_limiter.is_available()

  def get_order_stats(self):
    self._calc_conversion_ratio()
    return self._order_stats

  def _calc_conversion_ratio(self):
    for product in self._og_info.products:
      total = self._order_stats[product].num_submitted + self._order_stats[product].num_cancelled
      if total > 0:
        conversion_ratio = self._order_stats[product].num_fills / total
      else:
        conversion_ratio = 0
      self._order_stats[product].conversion_ratio = conversion_ratio

  @functools.lru_cache(maxsize=1024)
  def get_min_notional(self, product):
    if self._exchange_info is None:
      return 0

    for symbol_info in self._exchange_info['symbols']:
      if symbol_info['symbol'] == product.native_symbol:
        for filter_info in symbol_info['filters']:
          if filter_info['filterType'] == 'MIN_NOTIONAL':
            return float(filter_info['minNotional'])
    return None

  @tornado.gen.coroutine
  def _get_exchange_info(self):
    client = tornado.httpclient.AsyncHTTPClient()
    response = yield client.fetch('https://api.binance.com/api/v1/exchangeInfo')
    if response.code == requests.codes.ok:
      self._exchange_info = json.loads(response.body)
      self._scheduler.start()
      self._private_subscriber.start()
    else:
      self._logger.error('Failed to get exchange info!')
      self._ioloop.call_later(5, self._get_exchange_info)

  @tornado.gen.coroutine
  def _on_hit_order_rate_limit(self):
    if not self.is_ready():
      return

    self._scheduler.stop()
    self._og_ready.reset()
    self._logger.info('Hit rate limit, disable OG for % seconds.',
                      self._og_config.hit_rate_limit_cooldown_time)
    yield tornado.gen.sleep(self._og_config.hit_rate_limit_cooldown_time)
    self._scheduler.start()
    self._logger.info('Binance OG is reinitializing.')

  @tornado.gen.coroutine
  def _install_periodic_query(self):
    assert self._scheduler is not None, 'self._scheduler is not set!'
    assert self._publisher is not None, 'self._publisher is not set!'

    query_balance_task = wrap_async_task(
        self._private_client.query_account,
        msg_type=self._MsgType.REST_ACCOUNT_BALANCE,
        publisher=self._publisher,
    )

    # query balance
    self._scheduler.add_periodic_task(period=self._og_config.account_query_period,
                                      job_func=query_balance_task)

    # query active orders
    for product in self._og_info.products:
      query_orders_task = wrap_async_task(
          self._private_client.query_open_orders,
          msg_type=self._MsgType.OPEN_ORDERS,
          publisher=self._publisher,
          func_params=dict(symbol=product.native_symbol),
      )

      yield tornado.gen.sleep(self._og_config.burst_query_avoid_time)
      self._scheduler.add_periodic_task(
          period=self._og_config.orders_snapshot_query_period,
          job_func=query_orders_task,
      )

    # Query exchange system status.
    query_system_status_task = wrap_async_task(
        self._private_client.query_system_status,
        msg_type=self._MsgType.SYSTEM_STATUS,
        publisher=self._publisher,
    )

    self._scheduler.add_periodic_task(
        period=60,
        job_func=query_system_status_task,
    )

    def print_info():
      print_working_orders(self._logger, self)
      self._calc_conversion_ratio()
      headers = [
          'product',
          'num_submitted',
          'num_submitted_success',
          'num_cancelled',
          'num_cancelled_success',
          'num_fills',
          'disabled',
          'disabled_timestamp',
          'conversion_ratio'
      ]
      table = []
      for prod, stat in self._order_stats.items():
        if stat.disabled_timestamp:
          disabled_time = to_datetime(stat.disabled_timestamp).strftime('%Y%m%d %H%M%S')
        else:
          disabled_time = 'N/A'
        table.append([
            prod.symbol,
            stat.num_submitted,
            stat.num_submitted_success,
            stat.num_cancelled,
            stat.num_cancelled_success,
            stat.num_fills,
            stat.disabled,
            disabled_time,
            stat.conversion_ratio,
        ])
      fmt_table = tabulate.tabulate(table, headers, stralign="right")
      self._logger.info('----- Binance conversion ratio\n%s', fmt_table)

    # Periodically print working order list.
    self._scheduler.add_periodic_task(
        period=10,
        job_func=print_info,
    )

  def _async_submit_order(self, order, post_only):
    assert order.order_id is not None, order
    assert order.order_type == OrderType.LIMIT, order

    if self._order_stats[order.product].disabled:
      self._logger.warning('Product %s has low conversion ratio, it is disabled!')
      return

    min_notional = self.get_min_notional(order.product)
    pq = order.price * order.qty
    if pq < min_notional:
      self._order_manager.remove_order(order)
      self._logger.error('%s: price * qty (%s) < min_notional (%s)',
                         order.product.symbol,
                         pq,
                         min_notional)
      return

    client_order_id = self._to_client_order_id(order)
    formatter = self._order_formatter[order.product]
    binance_order_type = BinanceOrderType(order.order_type.name)
    if order.order_type == OrderType.LIMIT and post_only:
      binance_order_type = BinanceOrderType.LIMIT_MAKER
    params = {
        'symbol': order.product.native_symbol,
        'price': formatter.format_price(order.price),
        'quantity': formatter.format_qty(order.qty),
        'order_side': BinanceOrderSide(order.order_side.name),
        'order_type': binance_order_type,
        'new_client_order_id': client_order_id,
    }
    fut = self._private_client.place_order(**params)
    self._order_stats[order.product].num_submitted += 1
    return fut

  def _async_cancel_order(self, order):
    assert order.internal.exchange_order_id is not None, order
    fut = self._private_client.cancel_order(symbol=order.product.native_symbol,
                                            order_id=order.internal.exchange_order_id)
    self._order_stats[order.product].num_cancelled += 1
    return fut

  def _handle_submit_response(self, order, response):
    """
    {
    'clientOrderId': 'CLIENT_ORDER_ID',
    'cummulativeQuoteQty': '0.00000000',
    'executedQty': '0.00000000',
    'fills': [],
    'orderId': 78678287,
    'origQty': '40.00000000',
    'price': '0.00008200',
    'side': 'SELL',
    'status': 'NEW',
    'symbol': 'XRPBTC',
    'timeInForce': 'GTC',
    'transactTime': 1538043234760,
    'type': 'LIMIT'
    }

    // This error
    {'code': -1013, 'msg': 'Filter failure: PRICE_FILTER'}
    """
    msg = json.loads(response.content)
    if response.status_code != requests.codes.ok:
      self._logger.error('Order submission error: %s', msg)
      if order in self._order_manager:
        self._order_manager.remove_order(order, ignore_error=True)
      self._handle_error_response(response)
      return OrderEvent.ORDER_REJECTED

    self._order_stats[order.product].num_submitted_success += 1
    client_order_id = msg['clientOrderId']
    if not self._parse_client_order_id(client_order_id)[1] == order.order_id:
      self._logger.error('Weird! Wrong client_order_id: %s, %s', msg, order)
      self._order_manager.remove_order(order, ignore_error=True)
      return OrderEvent.ORDER_ERROR

    exchange_order_id = str(msg['orderId'])
    prev_order = self._order_manager.get(exchange_order_id=exchange_order_id)
    if prev_order is None:
      self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
    else:
      if prev_order.order_id != order.order_id:
        self._logger.info('order id replaced: %s => %s', prev_order.order_id, order.order_id)
        self._order_manager.remove_order(prev_order)
        self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
        self._order_manager.remove_order(order)
        prev_order.order_id = order.order_id
        self._order_manager.add_order(prev_order)

    status = msg.get('status', None)
    if status is not None:
      try:
        if BinanceOrderStatus(status) in (BinanceOrderStatus.PARTIALLY_FILLED,
                                          BinanceOrderStatus.FILLED):
          self._ioloop.add_callback(self._handle_fill_in_submit_resp, order, msg)
        elif BinanceOrderStatus(status) != BinanceOrderStatus.NEW:
          self._logger.info('submit response: %s', msg)
      except ValueError:
        self._logger.error('Unknown order status: %s', msg)
    else:
      self._logger.debug('Weird, status field is missing: %s', msg)
    return OrderEvent.ORDER_ACCEPTED

  def _handle_fill_in_submit_resp(self, order, msg):
    self._logger.info('handle fill info in submit response msg')
    avg_fill_price, qty_accum_fill, trade_ids = parse_fill_info_in_submit_resp(msg)
    last_fill = order.internal.fill.update_by_accum_fill(avg_fill_price, qty_accum_fill)
    if order.internal.fill.fully_filled:
      self._order_manager.set_order(order, fully_filled=True)

    self._publish_fill(last_fill.price_last_fill, last_fill.qty_last_fill, order)
    order.internal.fill.log_last_fill(order, self._logger)
    fill_type = coin_order_pb2.TAKER_FILL_TYPE
    self._og_logger.gen_order_filled_event_and_log(order, get_timestamp(), fill_type=fill_type)
    self._order_stats[order.product].num_fills += 1
    self._trade_ids.update(trade_ids)

  def _handle_cancel_response(self, order, response):
    """
    {
    'clientOrderId': 'CLIENT_ORDER_ID',
    'orderId': 78695355,
    'origClientOrderId': 'ORIG_CLIENT_ORDER_ID',
    'symbol': 'XRPBTC'
    }
    """
    msg = json.loads(response.content)
    if response.status_code != requests.codes.ok:
      self._logger.error('Cancel order error! %s', msg)
      self._handle_error_response(response)
      return OrderEvent.CANCEL_ERROR

    self._order_stats[order.product].num_cancelled_success += 1
    if not self._order_manager.is_dead_order(order):
      self._logger.debug('%s is cancel confirmed.', order.order_id)
      self._order_manager.set_order(order, cancel_confirmed=True)
    else:
      self._logger.debug('Order is dead already %s', order)
    return OrderEvent.CANCEL_CONFIRMED

  def _notification_callback(self, data):
    timestamp = get_timestamp()
    method_to_msg_type = {
        'executionReport': self._MsgType.ORDER_UPDATE,
        'outboundAccountInfo': self._MsgType.WS_ACCOUNT_BALANCE,
        'outboundAccountPosition': self._MsgType.WS_ACCOUNT_POSITION,
    }

    msg_type = method_to_msg_type.get(data['e'], None)
    if msg_type is None:
      self._logger.warning('Unknown event type: %s', data)
      return

    exchange_message = PrivateExchangeMessage(msg_type=msg_type,
                                              data=data,
                                              received_timestamp=timestamp,
                                              request_params=None,
                                              request_timestamp=None)
    self._publisher.write(timestamp, exchange_message)

  @exchange_message_handler(_MsgType.REST_ACCOUNT_BALANCE)
  def _handle_rest_account_balance(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    if response.status_code != requests.codes.ok:
      self._logger.error('Rest open orders response error! %s, %s',
                         response.status_code,
                         response.content)
      self._handle_error_response(response)
      return

    message = json.loads(response.content)
    self._og_info.update_balance_rest(message)
    self._og_logger.write_balance(self._og_info.balance_as_proto())
    if not self.is_ready():
      self._og_ready.set_balance_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.OPEN_ORDERS)
  def _handle_rest_open_orders(self, exchange_message: PrivateExchangeMessage):
    response = exchange_message.data
    if response.status_code != requests.codes.ok:
      self._logger.error('Rest open orders response error! %s, %s',
                         response.status_code,
                         response.content)
      self._handle_error_response(response)
      return

    timestamp = get_timestamp()
    message = json.loads(response.content)
    product = BinanceProduct.FromStrNativeProduct(exchange_message.request_params['symbol'])
    if product not in self._og_info.products:
      return

    exchange_open_order_dict = {}
    for order_msg in message:
      native_symbol = order_msg['symbol']
      product = BinanceProduct.FromStrNativeProduct(native_symbol)
      if product not in self._og_info.products:
        continue

      order = create_order_from_open_order_message(order_msg, timestamp, None)
      order_id = self._parse_client_order_id(order_msg['clientOrderId'])[1]
      if order_id is None:
        prev_order = self._order_manager.get(exchange_order_id=order.internal.exchange_order_id)
        if prev_order:
          order_id = prev_order.order_id
        else:
          order_id = self._order_id_generator.gen_order_id(is_foreign=True)
      order.order_id = order_id
      exchange_open_order_dict[order.internal.exchange_order_id] = order

    self._order_manager.copy_from_exchange_open_orders(exchange_open_order_dict, [product])

    if not self.is_ready():
      self._og_ready.set_orders_ready(True)
      self._og_ready.check_ready()

  @exchange_message_handler(_MsgType.WS_ACCOUNT_BALANCE)
  def _handle_ws_account_update(self, exchange_message: PrivateExchangeMessage):
    message = exchange_message.data
    self._og_info.update_balance_ws(message)
    self._og_logger.write_balance(self._og_info.balance_as_proto())

  @exchange_message_handler(_MsgType.WS_ACCOUNT_POSITION)
  def _handle_ws_account_position(self, exchange_message: PrivateExchangeMessage):
    pass

  @exchange_message_handler(_MsgType.ORDER_UPDATE)
  def _handle_ws_order_update(self, exchange_message: PrivateExchangeMessage):
    message = exchange_message.data
    timestamp = get_timestamp()
    execution_type = BinanceExecutionType(message['x'])
    if execution_type == BinanceExecutionType.CANCELED:
      client_order_id = message['C']
    else:
      client_order_id = message['c']

    trade_id = '%s_%s' % (message['s'], message['t'])
    if trade_id in self._trade_ids:
      return
    else:
      self._trade_ids.add(trade_id)

    order_id = self._parse_client_order_id(client_order_id)[1]
    recv_order = create_order_from_execution_report(message, order_id, timestamp)
    if recv_order.product not in self._og_info.products:
      self._logger.info('product not traded! %s', recv_order)
      return

    exchange_order_id = recv_order.internal.exchange_order_id
    work_order = self._order_manager.get(exchange_order_id=exchange_order_id)

    # In case a foreign order that has not been added yet
    if order_id is None and work_order is None:
      order_id = self._order_id_generator.gen_order_id(is_foreign=True)
      recv_order.order_id = order_id
      self._order_manager.add_order(recv_order)
      self._logger.info('Add foreign order: %s', recv_order)

    order = self._order_manager.get(exchange_order_id=exchange_order_id)
    if order is None:
      order = self._order_manager.get(order_id=recv_order.order_id)
      if order is None:
        self._order_manager.add_order(recv_order)
        order = self._order_manager.get(order_id=recv_order.order_id)

    if order:
      try:
        assert_orders_equivalent(order, recv_order, rel_tol=1e-4)
      except AssertionError:
        self._logger.warning('assertion order equivalent failed! %s\n%s', order, recv_order)
    else:
      self._logger.error('Unknown order %s\n%s', recv_order, message)

    if execution_type == BinanceExecutionType.NEW:
      if not order.accepted:
        self._order_manager.set_order(order, exchange_order_id=exchange_order_id)
      else:
        if order.internal.exchange_order_id != exchange_order_id:
          self._logger.error('Wrong exchange order id!\n%s\n%s', order, recv_order)
    elif execution_type == BinanceExecutionType.CANCELED:
      if self._order_manager.is_dead_order(order):
        self._logger.debug('Order (%s) is dead already, no need to set cancel_confirmed!',
                           order.order_id)
      else:
        self._order_manager.set_order(order, cancel_confirmed=True)
    elif execution_type == BinanceExecutionType.REJECTED:
      self._logger.info('Order rejected!, %s, %s', order, message)
      if order in self._order_manager:
        self._order_manager.remove_order(order, ignore_error=True)
    elif execution_type == BinanceExecutionType.TRADE:
      last_fill = order.internal.fill.update_by_accum_fill(
          recv_order.internal.fill.value.price_avg_fill,
          recv_order.internal.fill.value.qty_accum_fill)
      if order.internal.fill.fully_filled:
        self._order_manager.set_order(order, fully_filled=True)

      if last_fill is None:
        self._logger.warning('Weird order status! %s, %s', json.dumps(message), order)
        return

      self._publish_fill(last_fill.price_last_fill, last_fill.qty_last_fill, order)
      order.internal.fill.log_last_fill(order, self._logger)

      if message['m'] is True:
        fill_type = coin_order_pb2.MAKER_FILL_TYPE
      elif message['m'] is False:
        fill_type = coin_order_pb2.TAKER_FILL_TYPE
      else:
        fill_type = None
        self._logger.error('message[\'m\']: %s', message['m'])

      self._og_logger.gen_order_filled_event_and_log(order, timestamp, fill_type=fill_type)
      self._order_stats[order.product].num_fills += 1
    elif execution_type == BinanceExecutionType.REPLACED:
      self._logger.error('REPLACED execution type is not used!')
    else:
      self._logger.error('Invalid execution_type: %s', execution_type)

  @exchange_message_handler(_MsgType.ACCOUNT_STATUS)
  def _handle_account_status(self, exchange_message: PrivateExchangeMessage):
    try:
      message = json.loads(exchange_message.data.content)
    except ValueError:
      self._logger.error(
          'Account status %s, %s',
          exchange_message.data.status_code,
          exchange_message.data.content,
      )
      return

    self._logger.info('Account status: %s', message)
    success = message.get('success')
    msg = message.get('msg')
    if (not success) or (msg == 'Normal'):
      return

  @exchange_message_handler(_MsgType.SYSTEM_STATUS)
  def _handle_system_status(self, exchange_message: PrivateExchangeMessage):
    try:
      message = json.loads(exchange_message.data.content)
    except ValueError:
      self._logger.error(
          'system status response error: %s, %s',
          exchange_message.data.status_code,
          exchange_message.data.content,
      )
      return

    status = message.get('status')
    if status is None:
      return
    elif status == 0:
      self._logger.info('Exchange system status: normal. %s', message)
    else:
      self._logger.error('Exchange system status: maintenance. %s. OG will be disabled!', message)
      self._disable_og()
